package org.tao.lightningmq.common.event;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.event.model.BaseEvent;
import org.tao.lightningmq.common.utils.ReflectUtil;

import java.util.*;
import java.util.concurrent.*;

/**
 * @Author lht
 * @date 2025/1/7 - 19:38
 * @description:
 */
public class EventBus {

    private static final Logger LOGGER = LoggerFactory.getLogger(EventBus.class);

    private final Map<Class<? extends BaseEvent>, List<Listener>> eventListenerMap = new ConcurrentHashMap<>();
    private static String taskName = "event-bus-task-";
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10,
            100,
            3,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1000),
            r -> {
                Thread thread = new Thread(r);
                thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                    @Override
                    public void uncaughtException(Thread t, Throwable e) {
                        e.printStackTrace();
                    }
                });
                thread.setName(taskName + UUID.randomUUID());
                return thread;
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    public EventBus() {
    }

    public EventBus(String taskName) {
        EventBus.taskName = taskName;
    }

    public void init() {
        LOGGER.info("start init event bus");
        ServiceLoader.load(Listener.class).forEach(listener -> {
            Class clazz = ReflectUtil.getInterfaceT(listener, 0);
            this.registry(clazz, listener);
            LOGGER.debug("registry listener:{}", listener.getClass().getName());
        });
    }

    private <E extends BaseEvent> void registry(Class<? extends BaseEvent> clazz, Listener<E> listener) {
        List<Listener> listeners = eventListenerMap.computeIfAbsent(clazz, k -> new ArrayList<>());
        listeners.add(listener);
    }

    public void publish(BaseEvent event) {
        List<Listener> listeners = eventListenerMap.get(event.getClass());
        threadPoolExecutor.execute(() -> {
            try {
                for (Listener listener : listeners) {
                    listener.onReceive(event);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    public Listener getListenerByClass(Class<? extends BaseEvent> clz) {
        return this.eventListenerMap.get(clz).get(0);
    }

}
